package cn.ilikes.tools.nosql.jpa.clients.handers.hbases;

import java.io.IOException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.slf4j.Logger;

import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.nosql.jpa.JPASQLPersistenUnitEnv;
import cn.ilikes.tools.nosql.jpa.SysConfigs;
import cn.ilikes.tools.nosql.jpa.clients.dataSources.hbases.HbaseDataSource;
import cn.ilikes.tools.nosql.jpa.clients.handers.JPASQLDataHnaderException;
import cn.ilikes.tools.nosql.jpa.graphs.Node;
import cn.ilikes.tools.nosql.jpa.graphs.Node.NodeLink;
import cn.ilikes.tools.nosql.jpa.metadatas.EntityMetaData;
import cn.ilikes.tools.nosql.jpa.metadatas.elements.EntityFields;
import cn.ilikes.tools.nosql.jpa.metadatas.elements.Relation;
import cn.ilikes.tools.nosql.jpa.unitls.JPAUtils;
import cn.ilikes.tools.nosql.jpa.unitls.PropertyAccessorHelper;

public class HbaseServices {

	private Logger logger = Loggerfactory.instance(HbaseServices.class);

	private Map<String, HTableInterface> tableInterfaces = new HashMap<String, HTableInterface>();

	private HbaseDataSource dataSource;

	private JPASQLPersistenUnitEnv persistenUnitEnv;

	public HbaseServices(HbaseDataSource dataSource, JPASQLPersistenUnitEnv persistenUnitEnv) {
		this.dataSource = dataSource;
		this.persistenUnitEnv = persistenUnitEnv;
	}

	private HbaseDataSource getDataSource() {
		return dataSource;
	}

	private JPASQLPersistenUnitEnv getPersistenUnitEnv() {
		return persistenUnitEnv;
	}

	private HTableInterface getHbaseHtable(EntityMetaData entityMetaData) {
		String entityTableName = entityMetaData.getEntityTableName();
		HTableInterface hTableInterface = tableInterfaces.get(entityTableName);
		if (hTableInterface == null) {
			HbaseDataSource hbaseDataSource = getDataSource();
			HTablePool tablePool = hbaseDataSource.getDataSource();
			hTableInterface = tablePool.getTable(entityTableName);

			tableInterfaces.put(entityTableName, hTableInterface);
		}

		return hTableInterface;
	}

	public void commits() {
		Loggerfactory.info(logger, "Hbase  Services commits ");
		for (HTableInterface tableInterface : tableInterfaces.values()) {
			try {
				tableInterface.flushCommits();

			} catch (IOException e) {
				Loggerfactory.error(logger, e.getMessage(), e);
			}
		}
	}

	public void remove(Node node) throws JPASQLDataHnaderException {

		EntityMetaData entityMetaData = node.getEntityMetaData();
		Object entity = node.getEntity();
		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "remove  entity  : " + entity + " entity id : "
						+ PropertyAccessorHelper.fieldValue(entity, entityMetaData.getEntityId().getField()));
			HTableInterface hTableInterface = getHbaseHtable(entityMetaData);
			byte[] idKey = HBaseUtils.putToPutByte(entity, entityMetaData.getEntityId().getField());
			Delete delete = new Delete(idKey);
			hTableInterface.delete(delete);
		} catch (IOException e) {
			throw new JPASQLDataHnaderException(e);
		}

	}

	public Result find(EntityMetaData entityMetaData, Object key) throws JPASQLDataHnaderException {
		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "find  entity   key : " + key + "  entity class : " + entityMetaData.getEntityClazz());
			HTableInterface hTableInterface = getHbaseHtable(entityMetaData);
			Get get = new Get(HBaseUtils.putToPutByte(key));
			Result result = hTableInterface.get(get);
			return result;
		} catch (IOException e) {
			throw new JPASQLDataHnaderException(e);
		}
	}

	//EntityFieldsType RealocalKey   Relation  : true   jPAField is  ColumnName  :  false     columnName  :    null     Relation  ColumnName  :  PHOTO_ID
	//EntityFieldsType RealocalKey   Relation  : false   jPAField is  ColumnName  :  true     columnName  :    ALBUM_NAME     Relation  ColumnName  :  null
	//EntityFieldsType RoreignKey   Relation  : true   jPAField is  ColumnName  :  false     columnName  :    null     Relation  ColumnName  :  ALBUM_ID

	public void save(Node node) throws JPASQLDataHnaderException {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, " save  entity  : " + node.getEntityMetaData().getEntityTableName() + "  " + "  entity id :"
					+ PropertyAccessorHelper.fieldValue(node.getEntity(), node.getEntityMetaData().getEntityId().getField()));
		JPASQLPersistenUnitEnv persistenUnitEnv = getPersistenUnitEnv();
		HTableInterface hTableInterface = getHbaseHtable(node.getEntityMetaData());
		byte[] idKey = HBaseUtils.putToPutByte(node.getEntity(), node.getEntityMetaData().getEntityId().getField());
		try {
			Put put = new Put(idKey);
			Set<Map.Entry<String, EntityFields>> entityFields = node.getEntityMetaData().getEntityFields().entrySet();

			for (Entry<String, EntityFields> entityField : entityFields) {
				{
					EntityFields entityFields_ = entityField.getValue();
					//if (Loggerfactory.isdebug(logger))System.out.println(entityFields_.getFieldName()+"======11111 ======  "+entityFields_.isSave()+"    "+entityFields_.getjPAField().isRelation());
					if (entityFields_.getjPAField() == null || entityFields_.isSave() == false)
						continue;

					switch (entityFields_.getEntityFieldsType()) {
					case RealocalKey: {
						//System.out.println("============  "+entityFields_.getFieldName()+"    "+entityFields_.getjPAField().isRelation());
						if (entityFields_.getjPAField().isRelation()) {//有关系
							//有节点链接类型
							if (node.getNodeLink().size() > 0) {
								NodeLink nodeLink = node.getNodeLink().get(entityFields_);
								if (nodeLink == null)
									continue;
								Object value = nodeLink.getTargetObject();
								if (value == null)
									continue;
								Relation relation = entityFields_.getjPAField().getRelation();
								Class<?> targetEntity = relation.getTargetEntity();
								EntityMetaData targetEntityMetaData = persistenUnitEnv.getMetaDataManagerFactory().selectIEntityMetaData(targetEntity);
								EntityFields targetEntityFieldsid = targetEntityMetaData.getEntityId();

								Object id = PropertyAccessorHelper.getObject(value, targetEntityFieldsid.getField());
								String columnName = relation.getColumnName();
								int index = columnName.indexOf(SysConfigs.hbase_column_family_qualifier);
								if (Loggerfactory.isdebug(logger))
									Loggerfactory.debug(logger, "save entity realocal key relation column name  :  " + columnName + "  value : "
											+ id);
								put.add(HBaseUtils.getBytes(JPAUtils.getfamily(columnName, index)), HBaseUtils.getBytes(JPAUtils.getQualifier(
										columnName, index)), HBaseUtils.getBytes(id));

							}

						} else {//无关系

							Object value = PropertyAccessorHelper.getObject(node.getEntity(), entityFields_.getField());
							if (value == null)
								continue;

							String columnName = entityFields_.getjPAField().getColumnName();
							if (Loggerfactory.isdebug(logger))
								Loggerfactory.debug(logger, "save entity field column name  :  " + columnName + "  value : " + value);
							int index = columnName.indexOf(SysConfigs.hbase_column_family_qualifier);
							put.add(HBaseUtils.getBytes(JPAUtils.getfamily(columnName, index)),
									HBaseUtils.getBytes(JPAUtils.getQualifier(columnName, index)), HBaseUtils.getBytes(value));
						}

						break;
					}

					case RoreignKey: {
						//System.out.println("======RoreignKey ======  "+entityFields_.getFieldName()+"    "+entityFields_.getjPAField().isRelation());
						Relation relation = entityFields_.getjPAField().getRelation();
						Class<?> targetEntity = relation.getTargetEntity();
						EntityMetaData targetEntityMetaData = persistenUnitEnv.getMetaDataManagerFactory().selectIEntityMetaData(targetEntity);
						NodeLink nodeLink = node.getNodeLink().get(entityFields_);
						Object id = PropertyAccessorHelper.getObject(nodeLink.getTargetObject(), targetEntityMetaData.getEntityId().getField());
						String columnName = relation.getColumnName();
						int index = columnName.indexOf(SysConfigs.hbase_column_family_qualifier);
						if (Loggerfactory.isdebug(logger))
							Loggerfactory.debug(logger, "save entity roreign key relation column name  :  " + columnName + "  value : " + id);
						put.add(HBaseUtils.getBytes(JPAUtils.getfamily(columnName, index)), null, HBaseUtils.getBytes(id));
					}
					default:
					}
				}
			}
			hTableInterface.put(put);
		} catch (Exception e) {
			throw new JPASQLDataHnaderException(e);
		}
	}

	public Result[] seach(EntityMetaData entityMetaData, List<?> keyIds) throws JPASQLDataHnaderException {
		try {
			String entityTableName = entityMetaData.getEntityTableName();
			HbaseDataSource hbaseDataSource = getDataSource();
			HTableInterface hTableInterface = hbaseDataSource.getDataSource().getTable(entityTableName);
			List<Get> gets = new ArrayList<Get>();
			for (Object keyId : keyIds) {
				Get get = new Get(HBaseUtils.getBytes(keyId));
				gets.add(get);
			}
			Result[] results = hTableInterface.get(gets);
			return results;
		} catch (IOException e) {
			throw new JPASQLDataHnaderException(e);
		}

	}

	/**
	 * PooledHTable{,
	 * table=org.apache.hadoop.hbase.client.HTable@2052d1a4}
	 * 
	 * @param entityMetaData
	 * @param key
	 * @return
	 * @throws IOException
	 */
	public Result getObjectById(EntityMetaData entityMetaData, Object key) throws IOException {
		HTableInterface hTableInterface = getHbaseHtable(entityMetaData);
		Get get = new Get(HBaseUtils.getBytes(key));
		Result result = hTableInterface.get(get);
		return result;
	}

	public byte[] getfamilyAndqualifierValues(EntityMetaData entityMetaData, Object key, String family, String qualifier) throws IOException {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "find  Id   : " + key.getClass() + "  family : " + family + "    qualifier " + qualifier);
		HTableInterface hTableInterface = getHbaseHtable(entityMetaData);
		Get get = new Get(HBaseUtils.putToPutByte(key));
		Result result = hTableInterface.get(get);
		return result.getValue(HBaseUtils.getBytes(family), HBaseUtils.getBytes(qualifier));
	}

	public void update(Node node) throws JPASQLDataHnaderException {
		this.remove(node);
		this.save(node);

	}

	public byte[] findValue(EntityMetaData entityMetaData, Object key, String family, String qualifier) throws JPASQLDataHnaderException {
		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "find  Value  : " + key + "  family : " + family + "    qualifier " + qualifier);
			HTableInterface hTableInterface = getHbaseHtable(entityMetaData);

			Get get = new Get(HBaseUtils.getBytes(key));
			Result result = hTableInterface.get(get);
			return result.getValue(HBaseUtils.getBytes(family), HBaseUtils.getBytes(qualifier));
		} catch (IOException e) {
			throw new JPASQLDataHnaderException(e);
		}
	}
}